package org.lds.message.center.helloworld;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.lds.message.center.common.AbstractMessageService;
import org.lds.message.center.common.MessageHandler;
import org.lds.message.center.common.model.request.HelloWorldRequest;
import org.lds.message.center.common.util.AssertUtil;
import org.lds.message.center.helloworld.constant.HelloWorldConstant;
import org.lds.message.center.helloworld.enums.HelloWorldErrorEnum;
import org.lds.message.center.helloworld.properties.HelloWorldProperties;
import org.lds.message.center.plugin.core.kafka.KafkaProducer;

import java.util.Objects;

@Slf4j
public class HelloWorldMessageService extends AbstractMessageService<HelloWorldRequest> {
    private final HelloWorldProperties helloWorldProperties;
    private final KafkaProducer kafkaProducer;

    public HelloWorldMessageService(MessageHandler<HelloWorldRequest> messageHandler, HelloWorldProperties helloWorldProperties, KafkaProducer kafkaProducer) {
        super(messageHandler);
        this.helloWorldProperties = helloWorldProperties;
        this.kafkaProducer = kafkaProducer;
    }

    @Override
    public String channelName() {
        return HelloWorldConstant.CHANNEL_NAME;
    }

    @Override
    public String channelType() {
        return HelloWorldConstant.CHANNEL_TYPE;
    }

    @Override
    public Class<HelloWorldRequest> supportClass() {
        return HelloWorldRequest.class;
    }

    @Override
    public void check(HelloWorldRequest request) {
        AssertUtil.assertTrue(helloWorldProperties.isEnabled(), HelloWorldErrorEnum.PLUGIN_CLOSED);
        AssertUtil.assertTrue(Objects.nonNull(request), HelloWorldErrorEnum.REQUEST_INVALID);

        AssertUtil.assertTrue(StringUtils.isNotEmpty(request.getText()), HelloWorldErrorEnum.TEXT_INVALID);
    }

    @Override
    protected void store(HelloWorldRequest request) {
        kafkaProducer.send(helloWorldProperties.getTopic(), request);
    }

}
